package com.cennavi.config;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Properties;
import java.util.UUID;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月24日 下午5:11:19 
    * @version 1.0 
    * 读取kafka配置文件
    */

public class KafkaConfigFile {

	public String metadata_broker_list;
	public Properties props = null;
	public HashMap<String, String> kafkaParams = null;
	public String topics = null;
	public String outputTopic = null;
	public HashSet<String> topicsSet = null;

	public KafkaConfigFile() {

	}

	public void init(String kafkaConfigFilePath){
		props = new Properties();
		
		try {
			props.load(new FileInputStream(kafkaConfigFilePath));
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}

		topics = props.getProperty("kafka.topics");
		topicsSet = new HashSet<String>(Arrays.asList(topics.split(",")));
		outputTopic = props.getProperty("kafka.outPutTopic");
		metadata_broker_list = props.getProperty("metadata.broker.list");
		kafkaParams = new HashMap<String, String>();
		kafkaParams.put("zookeeper.connect", props.getProperty("zookeeper.connect"));
		kafkaParams.put("group.id", UUID.randomUUID().toString());
		kafkaParams.put("auto.offset.reset", props.getProperty("auto.offset.reset"));
		kafkaParams.put("metadata.broker.list", props.getProperty("metadata.broker.list"));
		//kafkaParams.put("bootstrap.servers", props.getProperty("bootstrap.servers"));
		kafkaParams.put("fetch.message.max.bytes", props.getProperty("fetch.message.max.bytes"));

	}
	
	public String getMetadata_broker_list() {
		return metadata_broker_list;
	}

	public void setMetadata_broker_list(String metadata_broker_list) {
		this.metadata_broker_list = metadata_broker_list;
	}

	public Properties getProps() {
		return props;
	}

	public void setProps(Properties props) {
		this.props = props;
	}

	public HashMap<String, String> getKafkaParams() {
		return kafkaParams;
	}

	public void setKafkaParams(HashMap<String, String> kafkaParams) {
		this.kafkaParams = kafkaParams;
	}

	public String getTopics() {
		return topics;
	}

	public void setTopics(String topics) {
		this.topics = topics;
	}

	public HashSet<String> getTopicsSet() {
		return topicsSet;
	}

	public void setTopicsSet(HashSet<String> topicsSet) {
		this.topicsSet = topicsSet;
	}

	public String getOutputTopic() {
		return outputTopic;
	}

	public void setOutputTopic(String outputTopic) {
		this.outputTopic = outputTopic;
	}
	
	
}
